
/**
 * The Merge filter combines the two separate input byte stream to a single
 * stream based on value of the time field
 * 
 * @author Team 10
 * 
 */
public class MergeFilter extends FilterFramework2 {

	public MergeFilter() {
		super(2, 1);
	}

	public void run() {
	
		boolean measureFlag1 = false;
		boolean measureFlag2 = false;

		// measurment1 object reads from source 1
		Measurement measurement1 = new Measurement();
		// measurment2 object reads from source 2
		Measurement measurement2 = new Measurement();
		try {
			// Reads the input pipe stream
			measurement1.readByteStream(this, 0);

			measurement2.readByteStream(this, 1);

		} catch (EndOfStreamException e1) {

			e1.printStackTrace();

		}

		// The comparator has a compare method to compare the two date values
		MeasurementComparator comparator = new MeasurementComparator();
		while (true) {
			/*************************************************************
			 * Here we read a byte and write a byte
			 *************************************************************/

			try {
				/*
				 * If measurement1 is greater than measurment2 write
				 * measurement1 to the outstream and read next value from input
				 * pipe 1
				 */
				if (comparator.compare(measurement1, measurement2) == 1) {
					measurement2.writeByteStream(this, 0);
					measurement2 = new Measurement();
					measureFlag2 = false;
					measurement2.readByteStream(this, 1);
					measureFlag2 = true;
				}/*
				 * If measurement1 is lesser then measurment2 write measurment2
				 * to the outstream and read next value from input pipe 2
				 */
				else if (comparator.compare(measurement1, measurement2) == -1) {
					measurement1.writeByteStream(this, 0);
					measurement1 = new Measurement();
					measureFlag1 = false;
					measurement1.readByteStream(this, 0);
					measureFlag1 = true;
				} else {
					/*
					 * If measurement1 is equal to measurment2 write measurment1
					 * and measurement 2 to the outstream and read next values
					 * from input pipe 1 and 2
					 */
					measurement1.writeByteStream(this, 0);
					measurement1 = new Measurement();
					measurement1.readByteStream(this, 0);
					measurement2.writeByteStream(this, 0);
					measurement2 = new Measurement();
					measurement2.readByteStream(this, 1);
				}

			}
			// Break out of loop when one input stream reaches the end of the
			// stream
			catch (EndOfStreamException e) {
				break;

			}

		}
		// Continue writing to the output pipe the values from the remaining
		// stream
		while (true) {
			try {
				if (measureFlag1 == false && measureFlag2 == true) {
					measurement2.writeByteStream(this, 0);
					measurement2.readByteStream(this, 1);
				} else {
					measurement1.writeByteStream(this, 0);
					measurement1.readByteStream(this, 0);
				}
			} catch (EndOfStreamException e) {
				break;

			} // catch

		}
		//Close all input ports
		CloseInputPorts(0);

	} // run

}
